package org.mq.mymq.broker.slave;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.mq.mymq.broker.event.model.CreateTopicEvent;
import org.mq.mymq.broker.event.model.PushMsgEvent;
import org.mq.mymq.common.cache.BrokerServerSyncFutureManager;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.CreateTopicReqDTO;
import org.mq.mymq.common.dto.MessageDTO;
import org.mq.mymq.common.dto.SlaveSyncRespDTO;
import org.mq.mymq.common.dto.StartSyncRespDTO;
import org.mq.mymq.common.enums.BrokerEventCode;
import org.mq.mymq.common.enums.BrokerResponseCode;
import org.mq.mymq.common.event.EventBus;
import org.mq.mymq.common.event.model.Event;
import org.mq.mymq.common.remote.SyncFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @ClassName SlaveSyncServerHandler
 * @Description Slave节点接收master节点数据信息handler
 * 1. 接收master节点创建topic信息，并发布创建topic事件
 * 2. 接收master节点的同步消息信息，并发布接收消息推送事件
 * 3. 接收master节点将Slave节点添加入同步消息缓存信息
 * @Author jiarong_ye
 * @Date 2025/7/11 17:37
 * @Version 1.0
 */
@ChannelHandler.Sharable
public class SlaveSyncServerHandler extends SimpleChannelInboundHandler<TcpMsg> {
    private static final Logger log = LoggerFactory.getLogger(BrokerEventCode.class);
    private EventBus eventBus;

    public SlaveSyncServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
        eventBus.init();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        Event event = null;
        if (BrokerEventCode.CREATE_TOPIC.getCode() == code) {
            CreateTopicReqDTO createTopicReqDTO = JSON.parseObject(body, CreateTopicReqDTO.class);
            CreateTopicEvent createTopicEvent = new CreateTopicEvent();
            createTopicEvent.setCreateTopicReqDTO(createTopicReqDTO);
            createTopicEvent.setMsgId(createTopicReqDTO.getMsgId());
            event = createTopicEvent;
        } else if (BrokerEventCode.PUSH_MSG.getCode() == code) {
            MessageDTO messageDTO = JSON.parseObject(body, MessageDTO.class);
            PushMsgEvent pushMsgEvent = new PushMsgEvent();
            pushMsgEvent.setMessageDTO(messageDTO);
            log.info("收到消息推送内容:{},message is {}", new String(messageDTO.getBody()), JSON.toJSONString(messageDTO));
            event = pushMsgEvent;
        } else if (BrokerResponseCode.START_SYNC_SUCCESS.getCode() == code) {
            StartSyncRespDTO startSyncRespDTO = JSON.parseObject(body, StartSyncRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManager.get(startSyncRespDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(tcpMsg);
            }
            return;
        } else if(BrokerResponseCode.SLAVE_SYNC_RESP.getCode() == code) {
            SlaveSyncRespDTO slaveSyncRespDTO = JSON.parseObject(body, SlaveSyncRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManager.get(slaveSyncRespDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(tcpMsg);
            }
        }
        event.setChannelHandlerContext(channelHandlerContext);
        eventBus.publish(event);
    }
}
